package tk.xboot.kfk;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.*;

import java.io.IOException;
import java.io.InputStream;
import java.time.LocalDate;
import java.util.*;
import java.util.stream.Collectors;

public class FirmOfferRatioETL {
    private static volatile boolean RUN = true;
    private static ObjectMapper mapper = new ObjectMapper();
    // 3个月到15天数据的取数频次
    private static int mod = 24 / 6;

    public static void main(String[] args) throws IOException {

        Properties propsSender = new Properties();
        InputStream inSender = DataProducer.class.getClassLoader().getResourceAsStream("producer.conf");
        propsSender.load(inSender);
        Producer<String, String> sender = new KafkaProducer<>(propsSender);


        Properties props = new Properties();
        InputStream in = DataConsumer.class.getClassLoader().getResourceAsStream("consumer.conf");
        props.load(in);


        final String SOURCE_TOPIC = props.getProperty("topic");
        final String SINK_TOPIC = propsSender.getProperty("topic");

        /* 定义consumer */
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        /* 消费者订阅的topic, 可同时订阅多个 */
        consumer.subscribe(Arrays.asList(SOURCE_TOPIC));


        Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
            public void run() {
                RUN = false;
            }
        });

        try {
            final long[] x = {0};
            while (RUN) {
                long cur = System.currentTimeMillis();
                ConsumerRecords<String, String> records = consumer.poll(10);
                for (ConsumerRecord<String, String> record : records){
                    String val = record.value();
                    //System.out.printf("get message from kafka:[ %s ]\n",val);
                    Map<String,Object> data = mapper.readValue(val, HashMap.class);

                    List<LinkedHashMap<String,Object>> list = (ArrayList<LinkedHashMap<String,Object>>)data.get("balanceModels");
                    data.put("balanceModels",reHash(list));

                    List<LinkedHashMap<String,Object>> list2 = (ArrayList<LinkedHashMap<String,Object>>)data.get("benifitModels");
                    data.put("benifitModels",reHash(list2));

                    List<LinkedHashMap<String,Object>> list3 = (ArrayList<LinkedHashMap<String,Object>>)data.get("incomeModels");
                    data.put("incomeModels",reHash(list3));

                    List<LinkedHashMap<String,Object>> list4 = (ArrayList<LinkedHashMap<String,Object>>)data.get("monthRatioModels");
                    //data.put("monthRatioModels",reHash(list4));
                    data.put("monthRatioModels",Collections.EMPTY_LIST);

                    List<LinkedHashMap<String,Object>> list5 = (ArrayList<LinkedHashMap<String,Object>>)data.get("weekRatioModels");
                    data.put("weekRatioModels",reHash(list5));
                    String newVal =  mapper.writeValueAsString(data);

                    sender.send(new ProducerRecord<String,String>(SINK_TOPIC, null, cur, null, newVal),
                            new Callback() {
                                @Override
                                public void onCompletion(RecordMetadata metadata, Exception exception) {

                                    if (exception != null) {
                                        System.out.println("Failed to send message with exception " + exception);
                                    } else {
                                        System.out.println("Send count:" + ++x[0]);
                                    }
                                }
                            });
                }
                consumer.commitAsync();
            }
        } catch (Throwable throwable) {
            System.out.println(throwable.getMessage());
            System.exit(1);
        }
    }


    // 计算3个月之前的数据,包含3个月当天的数据(- , latest 3]
    static LocalDate td3M = LocalDate.now().minusMonths(3);
    // 计算3个月到15天的数据，前开后闭,注意生成的数据的次序
    static LocalDate td15D = LocalDate.now().minusDays(15);

    private static List reHash(List<LinkedHashMap<String,Object>> list){
        if (list.isEmpty()) {
            return Collections.EMPTY_LIST;
        }

        LinkedHashMap<String, Object> first = list.get(0);

        // 1.  入盘超过3个月
        boolean comp3M = td3M.toString().compareTo(first.get("ctime").toString())>=0;
        if (comp3M) {
            // 入盘超过3个月,每天取一根
            return list.stream()
                    .collect(Collectors.groupingBy(e -> e.get("ctime")))
                    .entrySet().stream()
                    .sorted(Comparator.comparing(m -> m.getKey().toString()))
                    .collect(Collectors.toMap(Map.Entry::getKey,Map.Entry::getValue,(oldVal,newVal)->oldVal,LinkedHashMap::new))
                    .values().stream().map(o->o.get(0)).collect(Collectors.toList());
        }


        // 2.  入盘超过15天小于3个月,每天取6根()
        boolean comp3MonTo15D = td3M.toString().compareTo(first.get("ctime").toString())< 0
                && td15D.toString().compareTo(first.get("ctime").toString())>=0;
        if(comp3MonTo15D){
            // 每天取6根()
            List<LinkedHashMap<String,Object>> lst6H = new ArrayList<LinkedHashMap<String,Object>>();
            for (int i = 0; i < list.size(); i += mod) {
                lst6H.add(list.get(i));
            }
            return lst6H;
        }

        // 3. 15天内的直接返回
        return list;
    }

    private static List reFilter(List<LinkedHashMap<String,Object>> list){
        if (list.isEmpty()) {
            return Collections.EMPTY_LIST;
        }
        // 计算3个月之前的数据,包含3个月当天的数据(- , latest 3]
        LocalDate td3M = LocalDate.now().minusMonths(3);
        List<LinkedHashMap<String,Object>> lst3M = list.stream()
                .filter(o-> td3M.toString().compareTo(o.get("ctime").toString())>=0)
                .collect(Collectors.groupingBy(e -> e.get("ctime")))
                .entrySet().stream()
                .sorted(Comparator.comparing(m -> m.getKey().toString()))
                .collect(Collectors.toMap(Map.Entry::getKey,Map.Entry::getValue,(oldVal,newVal)->oldVal,LinkedHashMap::new))
                .values().stream().map(o->o.get(0)).collect(Collectors.toList());


        // 计算3个月到15天的数据，前开后闭,注意生成的数据的次序
        LocalDate td15D = LocalDate.now().minusDays(15);
        List<LinkedHashMap<String,Object>> lst3MonTo15D = list.stream()
                .filter(o->td3M.toString().compareTo(o.get("ctime").toString())< 0 && td15D.toString().compareTo(o.get("ctime").toString())>=0)
                .collect(Collectors.toList());

        // 每6小时取一根()
        List<LinkedHashMap<String,Object>> lst6H = new ArrayList<LinkedHashMap<String,Object>>();
        for (int i = 0; i < lst3MonTo15D.size(); i += mod) {
            lst6H.add(lst3MonTo15D.get(i));
        }
        // 计算15天内的数据，不含有15天的数据,注意生成的数据的次序
        List<LinkedHashMap<String,Object>> lst15D = list.stream()
                .filter(o->td15D.toString().compareTo(o.get("ctime").toString())<0)
                .collect(Collectors.toList());

        lst3M.addAll(lst6H);
        lst3M.addAll(lst15D);
        return lst3M;
    }
}
